In many applications, it is necessary for one process (computer program, module, or thread) executing on a computer system to communicate with one or more other processes executing on the same system or on other computer systems. One mechanism that facilitates process-to-process communication in a variety of systems is a “message queue”. To use a message queue, processes (“enqueue sessions”) send information to other processes (“dequeue sessions”) by placing messages in a message queue. The dequeue sessions obtain the information from the enqueue sessions by reading the messages from the message queue. When all dequeue sessions that need to read a given message from the message queue have read the given message, the message is removed from the message queue.
Unfortunately, implementations of message queues may not scale well. Specifically, as the number of dequeue sessions increases, the contention for the “hot” messages at the head of the message queue increases, thereby degrading performance. In addition, when the enqueue sessions and dequeue sessions are spread across several systems, the amount of communication on the interconnect between the systems can become excessive.
A database management system (DBMS) may implement sharded message queues as a way to allow scaling of messaging throughput within the system by horizontally partitioning queue stores across multiple logical message queue entities called “shards”. A DBMS may maintain dozens or hundreds of sharded message queues created by users of the database. A sharded message queue may have only one subscriber (e.g., a Java Message Service (“JMS”) queue), or it may have multiple subscribers (e.g., a JMS topic). In the latter case, the subscribers are independent and may dequeue messages at different positions in the sharded queue.
Generally, message queues are implemented as first-in first-out queues. However, a client may require asynchronous processing of messages, i.e., particular messages are to be made available for dequeue at particular user-indicated times in the future. For example, JMS 2.0 provides a “message visibility time” property for a given message, which allows the provider to make a message visible at a certain pre-specified time. Clients can specify this time using the JMS header property “JMSDeliveryTime”. JMS 2.0 also introduces a new message property called “delay”. This property limits the visibility of a given message until the delay time indicated by the property has elapsed.
JMS is an application programming interface (API) for sending messages between clients. JMS is a messaging standard that allows applications to create, send, receive, and read messages. Embodiments are described in the context of JMS, however embodiments may be implemented using any messaging mechanism, and are not limited to JMS.
In a system that is able to process delayed messages within message queues, the delivery time of each message in a queue is either explicitly indicated (e.g., via a “message visibility time” property value), or is determined based on the enqueue time of the message plus any indicated delay time for the message (e.g., via a “delay” property value). For non-delayed messages, the “delay” property or “message visibility time” property value for the message is generally null (or “0” as appropriate) and, as such, the delay time for the message is 0. Delayed messages are inserted into the message queue at the time of receipt (i.e., enqueue time), but are not immediately available for dequeue.
In some cases, systems avoid making delayed messages available for dequeue before the delivery time of the messages by placing delayed messages into a temporary staging area rather than into the message queue from which messages are being actively consumed. Such systems publish delayed messages (i.e., make them available for dequeue) by moving delayed messages from the temporary staging area into the message queue at the respective delivery times of the delayed messages, which makes the messages available for dequeue at approximately the delivery time of the message.
However, since delayed messages must be placed in the temporary staging area and then moved according to this technique, publishing a delayed message in this way is costly because it involves at least two insert operations and one delete operation over the database tables implementing the message queue. Furthermore, resources (such as a background timekeeper process) must be dedicated to perform the needed movement of delayed messages from the temporary staging area to the message queue at the correct time. There is some latency involved in moving the messages to the message queue, which generally causes delayed messages to be available at somewhat later times than their assigned delivery times indicate.
In other cases, systems avoid making delayed messages available for dequeue before the delivery time of the messages by putting all messages into the message queue, but maintaining delayed messages as invisible (i.e., based on a message state) until the message is eligible for consumption. Once a delayed message is eligible for consumption, the system marks the message as visible, which allows the message to be consumed by subscribers.
However, such techniques require updating the properties of delayed messages, to change the message state, at the respective delivery times of the messages in order to make the messages visible within the message queue. Again, this technique requires resources that are dedicated to performing the needed property updates for delayed messages at the respective delivery times. Furthermore, changing the visibility property of delayed messages introduces latency into the system such that delayed messages are generally made available at somewhat different times than their respective delivery times indicate.
Furthermore, such techniques generally store messages (both delayed and non-delayed) in the order in which the messages were received. Thus, delayed messages (interleaved with non-delayed messages) are delivered at later times than their neighboring non-delayed messages. This storage system significantly delays bulk deletion of messages since some of the messages in a given block are eligible for dequeue only after a delay. For example, messages in an example message queue are stored in a partitioned table having N partitions. When all of the messages in a given partition have been delivered to all subscribers, the partition is eligible for bulk deletion (such as using a truncate table operation). In a message queue with only non-delayed messages, such bulk deletion is an effective way to free up database space since the partitions empty out basically in order. However, if delayed messages are interspersed throughout all partitions in the table, bulk deletion of a given table partition is delayed until the delivery time of all of the delayed messages stored in the partition have passed and those messages have been delivered to all required subscribers.
It would be beneficial to allow delayed consumption of messages from a sharded message queue without requiring resources (such as a background timekeeper process) that are dedicated to detecting the time at which delayed messages should be made available. Also, it would be beneficial to make delayed messages available more precisely at the indicated delivery times of the messages. Furthermore, it would be beneficial to allow for efficient bulk deletion of both delayed and non-delayed messages from database storage.
The approaches described in this section are approaches that could be pursued, but not necessarily approaches that have been previously conceived or pursued. Therefore, unless otherwise indicated, it should not be assumed that any of the approaches described in this section qualify as prior art merely by virtue of their inclusion in this section.